package com.rrd.dw.mr.all;

import java.io.IOException;
import java.io.InputStream;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.google.gson.Gson;
import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.HBaseUtil;
import com.rrd.dw.utils.MD5Util;

/**
 * 
 * @author liubaoxin
 *
 */
public class ExportALLTaobaoDataToDW extends Configured implements Tool {
	public static final String CRT001 = "\001";
	public static final String[] TAOBAO_COLUMNS = { "nick_name", "real_name", "birthday", "sex", "login_name",
			"source_user_mobile", "source_user_idno", "source" };// taobao
	public static final String[] ORDER_COLUMNS = { "amount", "time:date", "order_no", "order_quantity", "freight",
			"order_status", "address" };// taobao

	public static final String[] ADDRESS_COLUMNS = { "consignee", "province", "city", "full_address", "post_code",
			"mobile" };// taobao

	public static final String[] PRODUCT_COLUMNS = { "shop_name", "original", "promotion", "name", "quantity:int",
			"attr", "service_info" };// taobao
	public static final String[] PAY_COLUMNS = { "mess_type", "red_pack", "zhifubao_amount", "bankcard_count",
			"huabei_limit", "jifenbao" };// pay

	public static final String[] JOB_COLUMNS = { "finished:boolean", "requested:boolean", "successful:boolean",
			"start_time:date", "end_time:date", "system_id", "query", "try_times:int", "type", "message", "code" };// job

	@Override
	public int run(String[] args) throws Exception {
		Properties prop = new Properties();
		try (InputStream is = this.getClass().getClassLoader().getResourceAsStream("hbase.properties");) {
			prop.load(is);
		}

		Job job = Job.getInstance(HBaseConfiguration.create(this.getConf()));
		Configuration conf = job.getConfiguration();
		conf.set("hbase.zookeeper.quorum", prop.get("hbase.zookeeper.quorum").toString());
		conf.set("hbase.rootdir", prop.get("hbase.rootdir").toString());

		conf.set("mapreduce.map.memory.mb", "3072");
		conf.set("mapreduce.reduce.memory.mb", "5120");
		conf.set("hbase.client.scanner.timeout.period", "120");
		conf.set("mapreduce.output.fileoutputformat.compress", "true");
		conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
		conf.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.GzipCodec");
		conf.set("io.compression.codecs", "org.apache.hadoop.io.compress.GzipCodec");
		conf.set("mapreduce.task.io.sort.factor", "10");
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if (otherArgs.length != 4) {
			System.err.println("Usage: DataExportTool <out> <start_time> <end_time>");
			return -1;
		}
		Long startTime = HBaseUtil.parseDate(otherArgs[1]).getTime();
		Long endTime = HBaseUtil.parseDate(otherArgs[2]).getTime();
		String model = otherArgs[3];
		job.setJarByClass(ExportALLTaobaoDataToDW.class);
		List<Scan> scans = new ArrayList<>();

		Scan scan = new Scan();
		scan.setCacheBlocks(false);
		scan.setCaching(200);
		scan.setMaxVersions(1);
		scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes("alitaobao_records"));
		if (!model.equals("ALL")) {
			System.out.println("增量运行.............");
			scan.setTimeRange(startTime, endTime + 14400000);
		} else {
			System.out.println("全量运行.............");

		}
		scans.add(scan);

		TableMapReduceUtil.initTableMapperJob(scans, ExportTaobaoDataToDWMapper.class, Text.class, Text.class, job);
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[0]));
		MultipleOutputs.addNamedOutput(job, "basepayinfo", TextOutputFormat.class, Text.class, Text.class);
		MultipleOutputs.addNamedOutput(job, "orderlist", TextOutputFormat.class, Text.class, Text.class);
		MultipleOutputs.addNamedOutput(job, "prodlist", TextOutputFormat.class, Text.class, Text.class);
		MultipleOutputs.addNamedOutput(job, "addresslist", TextOutputFormat.class, Text.class, Text.class);

		job.setReducerClass(ExportTaobaoDataToDWReducer.class);
		job.setNumReduceTasks(5);
		return job.waitForCompletion(true) ? 0 : 1;

	}

	public static void main(String[] args) throws Exception {
		Configuration conf = HBaseConfiguration.create();
		int res = ToolRunner.run(conf, new ExportALLTaobaoDataToDW(), args);
		System.exit(res);
	}

	public static class ExportTaobaoDataToDWMapper extends TableMapper<Text, Text> {

		@Override
		protected void map(ImmutableBytesWritable key, Result value, Context context)
				throws IOException, InterruptedException {
			Thread.sleep(20);
			StringBuffer sb = new StringBuffer("");
			Object[] rowkey = HBaseUtil.decodeRowkey(key.get());
			if (rowkey == null) {
				return;
			}
			long timestamp = Long.parseLong(rowkey[0].toString());
			String timestampstr = DateUtil.formateDateStr(new Date(timestamp));
			String jobid = rowkey[1].toString();
			Cell dataCell = null;
			Cell countCell = null;
			String type = "";
			String field = "";
			String user_id = MD5Util.getMd5(jobid + (timestampstr == null ? "" : timestampstr));
			sb.append(user_id).append(CRT001).append(jobid).append(CRT001).append(timestampstr).append(CRT001);
			// 淘宝基本信息
			for (String str1 : TAOBAO_COLUMNS) {
				String[] fieldArray = str1.split(":");
				if (fieldArray.length > 1) {
					field = fieldArray[0];
					type = fieldArray[1];
				} else {
					field = fieldArray[0];
					type = "string";
				}
				dataCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes(field));
				if (dataCell != null) {
					sb.append(HBaseUtil.getValue(type, CellUtil.cloneValue(dataCell)));
				}
				sb.append(CRT001);
			}
			// pay信息
			for (String str1 : PAY_COLUMNS) {
				String[] fieldArray = str1.split(":");
				if (fieldArray.length > 1) {
					field = fieldArray[0];
					type = fieldArray[1];
				} else {
					field = fieldArray[0];
					type = "string";
				}
				dataCell = value.getColumnLatestCell(Bytes.toBytes("pay"), Bytes.toBytes(field));
				if (dataCell != null) {
					sb.append(HBaseUtil.getValue(type, CellUtil.cloneValue(dataCell)));
				}
				sb.append(CRT001);
			}
			// job信息
			for (String str1 : JOB_COLUMNS) {
				String[] fieldArray = str1.split(":");
				if (fieldArray.length > 1) {
					field = fieldArray[0];
					type = fieldArray[1];
				} else {
					field = fieldArray[0];
					type = "string";
				}
				dataCell = value.getColumnLatestCell(Bytes.toBytes("job"), Bytes.toBytes(field));
				if (dataCell != null) {
					sb.append(HBaseUtil.getValue(type, CellUtil.cloneValue(dataCell)));
				}
				sb.append(CRT001);
			}
			String outString = sb.substring(0, sb.length() - 1).replace("\n", " ");
			context.write(new Text("basepayinfo"+user_id), new Text(outString));
			sb = null;
			// 订单信息
			countCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes("AliTaobaoOrdercount"));
			String order_id = "";
			if (countCell != null) {
				int count = Bytes.toInt(CellUtil.cloneValue(countCell));
				for (int i = 0; i < count; i++) {
					String orderNo = null;
					sb = new StringBuffer("");
					sb.append(user_id).append(CRT001).append(jobid).append(CRT001).append(timestampstr).append(CRT001);
					for (String bstr : ORDER_COLUMNS) {
						String[] fieldArray = bstr.split(":");
						if (fieldArray.length > 1) {
							field = fieldArray[0];
							type = fieldArray[1];
						} else {
							field = fieldArray[0];
							type = "string";
						}
						dataCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes(field + "#" + i));
						if (dataCell != null) {
							sb.append(HBaseUtil.getValue(type, CellUtil.cloneValue(dataCell)));
							if ("order_no".equals(bstr)) {
								orderNo = HBaseUtil.getValue(type, CellUtil.cloneValue(dataCell)).toString();
							}
						}
						sb.append(CRT001);
					}
					outString = sb.substring(0, sb.length() - 1).replace("\n", " ");
					order_id = MD5Util.getMd5(outString);
					outString = order_id + CRT001 + outString;
					context.write(new Text("orderlist" + jobid), new Text(outString));
					sb = null;

					// 商品
					countCell = value.getColumnLatestCell(Bytes.toBytes("taobao"),
							Bytes.toBytes("AliTaobaoOrderProductcount#" + i));
					if (countCell != null) {
						int pcount = Bytes.toInt(CellUtil.cloneValue(countCell));
						for (int j = 0; j < pcount; j++) {
							sb = new StringBuffer("");
							sb.append(user_id).append(CRT001).append(jobid).append(CRT001);
							sb.append(timestampstr).append(CRT001).append(orderNo).append(CRT001);
							for (String bstr : PRODUCT_COLUMNS) {
								String[] fieldArray = bstr.split(":");
								if (fieldArray.length > 1) {
									field = fieldArray[0];
									type = fieldArray[1];
								} else {
									field = fieldArray[0];
									type = "string";
								}
								dataCell = value.getColumnLatestCell(Bytes.toBytes("taobao"),
										Bytes.toBytes(field + "#" + i + "#" + j));
								if (dataCell != null) {
									sb.append(HBaseUtil.getValue(type, CellUtil.cloneValue(dataCell)));
								}
								sb.append(CRT001);
							}
							outString = sb.toString().substring(0, sb.length() - 1).replace("\n", " ");
							outString = MD5Util.getMd5(outString) + CRT001 + outString;
							context.write(new Text("prodlist"+ orderNo), new Text(outString));
							sb = null;
						}
					}
				}
			}
			// 地址信息
			countCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes("AliTaobaoAddresscount"));
			if (countCell != null) {
				int count = Bytes.toInt(CellUtil.cloneValue(countCell));
				for (int i = 0; i < count; i++) {
					sb = new StringBuffer("");
					sb.append(user_id).append(CRT001).append(jobid).append(CRT001);
					sb.append(timestampstr).append(CRT001);
					for (String bstr : ADDRESS_COLUMNS) {
						String[] fieldArray = bstr.split(":");
						if (fieldArray.length > 1) {
							field = fieldArray[0];
							type = fieldArray[1];
						} else {
							field = fieldArray[0];
							type = "string";
						}
						dataCell = value.getColumnLatestCell(Bytes.toBytes("taobao"), Bytes.toBytes(field + "#" + i));
						if (dataCell != null) {
							sb.append(HBaseUtil.getValue(type, CellUtil.cloneValue(dataCell)));
						}
						sb.append(CRT001);
					}
					outString = sb.toString().substring(0,sb.length()-1).replace("\n", " ");
					outString=MD5Util.getMd5(outString)+CRT001+outString;
					context.write(new Text("addresslist"  + jobid),
							new Text(outString));
					sb = null;
				}
			}
		}

	}

	public static class ExportTaobaoDataToDWReducer extends Reducer<Text, Text, Text, Text> {
		private MultipleOutputs<Text, Text> mos;
		private NullWritable k = NullWritable.get();
		@Override
		protected void setup(Context context) throws IOException, InterruptedException {
			this.mos = new MultipleOutputs<>(context);
		}

		@Override
		protected void cleanup(Context context) throws IOException, InterruptedException {
			this.mos.close();
		}

		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			String kstr = key.toString();
			for (Text val : values) {
				if (kstr.startsWith("basepayinfo")) {
					this.mos.write("basepayinfo", k, val);
				} else if (kstr.startsWith("prodlist")) {
					this.mos.write("prodlist", k, val);
				} else if (kstr.startsWith("addresslist")) {
					this.mos.write("addresslist", k, val);
				} else if(kstr.startsWith("orderlist")) {
					this.mos.write("orderlist", k, val);
				}
			}

		}
	}
}
